#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include"../mqcommon/mq_helper.hpp"
#include"../mqcommon/mq_logger.hpp"
#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_threadpool.hpp"

#include"mq_consumer.hpp"


 #include "muduo/proto/codec.h" 
#include "muduo/proto/dispatcher.h"
#include"muduo/net/TcpConnection.h"
#include<iostream>
#include<unordered_map>
#include<mutex>
#include<condition_variable>

namespace bitmq{
      using ProtobufCodecPtr=std::shared_ptr<ProtobufCodec>;
        using basicCommonResponsePtr=std::shared_ptr<basicCommonRespose>;
        using basicConsumerResposePtr=std::shared_ptr<basicConsumerRespose>;
    class Channel{
      
        public:
        using ptr=std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr& connection,const ProtobufCodecPtr&codec):_cid(UUIDHelper::uuid()),_connection(connection),_codec(codec){}
        ~Channel(){
            basicConcel();
        }
        std::string cid()
        {
            return _cid;
        }
        bool openChannel()
        {
               std::string rid=UUIDHelper::uuid();
                openChannelRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                _codec->send(_connection,req);
           
                basicCommonResponsePtr resp=waitbasicCommonRespose(rid);
                //返回
                return resp->ok();
        }
        void closeChannel(const std::string&cid)
        {
             std::string rid=UUIDHelper::uuid();
                closeChannelRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                _codec->send(_connection,req);
           
                waitbasicCommonRespose(rid);
                //返回
                return;
        }
        //声明交换机
        bool declareExchange(
            const std::string& ename,
            ExchageType type,
            bool durable,
            bool auto_delete,
             google::protobuf::Map<std::string, std::string>& args){
                //构造请求
                std::string rid=UUIDHelper::uuid();
                declareExchangeRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                req.set_type(type);
                req.set_durable(durable);
                req.set_auto_delete(auto_delete);
                req.mutable_args()->swap(args);
                //发送请求给服务端
                _codec->send(_connection,req);
                //等待响应   //为什么？moduo里面有发送缓冲区和接受缓冲区，再发送网路请求时，
                //和我们的主线程是异步的。为了保证我们的在创建交换机时，不被其他干扰，要保证服务器响应后。才可做其他请求
                basicCommonResponsePtr resp=waitbasicCommonRespose(rid);
                //返回
                return resp->ok();
            }
            //删除交换机
        void deleteExchange(const std::string& ename)
        {
              std::string rid=UUIDHelper::uuid();
                deleteExchangeRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_exchange_name(ename);
                _codec->send(_connection,req);
           
                waitbasicCommonRespose(rid);
                //返回
                return ;
        }
         //声明队列
        bool declareQueue(const std::string qname,
        bool durable,
        bool exclusive,
        bool auto_delete,
        google::protobuf::Map<std::string, std::string>& args)
        {
            std::string rid=UUIDHelper::uuid();
                declareQueueRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(qname);
                req.set_durable(durable);
                req.set_exclusive(exclusive);
                req.set_auto_delete(auto_delete);
                req.mutable_args()->swap(args);
                _codec->send(_connection,req);
           
                basicCommonResponsePtr resp=waitbasicCommonRespose(rid);
                //返回
                return resp->ok();
        }
        //删除队列
        void deleteQueue(const std::string qname)
        {
             std::string rid=UUIDHelper::uuid();
                deleteQueueRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(qname);
                _codec->send(_connection,req);
           
                waitbasicCommonRespose(rid);
                //返回
                return ;
        }
        //队列绑定
        bool basicBind(const std::string& ename,const std::string qname,const std::string&binding_key)
        {
             std::string rid=UUIDHelper::uuid();
                queueBindingRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(qname);
                req.set_exchange_name(ename);
                req.set_binding_key(binding_key);
                _codec->send(_connection,req);
           
                basicCommonResponsePtr resp=waitbasicCommonRespose(rid);
                //返回
                return resp->ok();
        }
        //队列解绑
        void basicUnBind(const std::string& ename,const std::string qname)
        {
                std::string rid=UUIDHelper::uuid();
                queueUnBindingRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(qname);
                req.set_exchange_name(ename);
               
                _codec->send(_connection,req);
           
                 waitbasicCommonRespose(rid);
                //返回
                return ;
        }
        //消息发布
        void basicPubish(const std::string&ename,bitmq::BasicProperties*bp,const std::string&body)
        {
              std::string rid=UUIDHelper::uuid();
                basicPublishRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_body(body);
                req.set_exchange_name(ename);
		std::cout << rid <<std::endl;
               if(bp)
               {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
               }
                _codec->send(_connection,req);
          	//std::cout << "消息发送成功" << std::endl; 
                 waitbasicCommonRespose(rid);
		// std::cout << " 收到对应响应" << std::endl;
                //返回
                return ;
        }
        //消息的确认
        void basicAck(const std::string&msg_id)
        {
            if(_consumer.get()==nullptr)
            {
                DLOG("消息确认时，找不到消费者信息");
                return;
            }
            std::string rid=UUIDHelper::uuid();
                basicAckRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(_consumer->_qname);
                req.set_msg_id(msg_id);

                   _codec->send(_connection,req);
           
                 waitbasicCommonRespose(rid);
                //返回
                return ;
        }
        //消息的订阅
        bool basicConsumer(const std::string&qname,const std::string&consumer_tag,bool auto_delete,const ConsumerCallBack&cb)
        {
            if(_consumer.get()!=nullptr)
            {
                DLOG("当前信道已经订阅其他队列消息！");
                return false;
            }
             std::string rid=UUIDHelper::uuid();
                basicConsumerRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_consumer_tag(consumer_tag);
                req.set_queue_name(qname);
                req.set_auto_ack(auto_delete);

                 _codec->send(_connection,req);
           
                 basicCommonResponsePtr resp=waitbasicCommonRespose(rid);
                 if(resp->ok()==false)
                 {
                    DLOG("发布消息失败!");
                    return false;
                 }
                
                 _consumer=std::make_shared<Consumer>(consumer_tag,qname,auto_delete,cb);
                return true;

        }
        //取消订阅
        void basicConcel()
        {
             if(_consumer.get()==nullptr)
            {
                return;
            }
                std::string rid=UUIDHelper::uuid();
                basicConcelRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(_consumer->_qname);
                req.set_consumer_tag(_consumer->_tag);

                 _codec->send(_connection,req);
           
                 waitbasicCommonRespose(rid);
                 _consumer.reset();
                //返回
                return ;
        }
        public:
        //连接收到基础响应后，向hash_map中添加响应
      
          void putbasicCommonResponse(const basicCommonResponsePtr&resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(),resp));
            _cv.notify_all();
        }
        ////连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进⾏消息处理
        void consume(const basicConsumerResposePtr&resp)
        {
            if(_consumer.get()==nullptr)
            {
                DLOG("消费者在处理消息时，未找到消费者信息!");
                return;
            }
            if(_consumer->_tag!=resp->consumer_tag())
            {
                DLOG("消费者标识不一致，无法处理消息!");
                return;
            }
            _consumer->_callback(resp->consumer_tag(),resp->mutable_properties(),resp->body());
        }
        private:
        basicCommonResponsePtr waitbasicCommonRespose(std::string rid)
        {
            //先加锁
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock,[&rid,this](){
                return _basic_resp.find(rid)!=_basic_resp.end();//while(condition) _cv.wait  //有响应了返回
            });
            basicCommonResponsePtr resp=_basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }
        private:
        std::string _cid;//信道id
        muduo::net::TcpConnectionPtr _connection;//moduo库的连接对象
        ProtobufCodecPtr _codec;//协议处理对象
        bitmq::Consumer::ptr _consumer;//消费者
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp;//信道id&响应对象
    };
    class ChannelManager
    {
        public:
        using ptr=std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        Channel::ptr create(const muduo::net::TcpConnectionPtr&connection,const ProtobufCodecPtr&codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto channel=std::make_shared<Channel>(connection,codec);
            channels.insert(std::make_pair(channel->cid(),channel));
            return channel;
        }
        void remove(const std::string&cid)
        {
             std::unique_lock<std::mutex> lock(_mutex);
             auto it=channels.find(cid);
             if(it==channels.end())
             {
                DLOG("client:删除信道时,没找到信道id!");
                return;
             }
             channels.erase(cid);
        }
        Channel::ptr get(const std::string&cid)
        {
             std::unique_lock<std::mutex> lock(_mutex);
             auto it=channels.find(cid);
             if(it==channels.end())
             {
                DLOG("client:获取信道时,没找到信道id!");
                return Channel::ptr();
             }
             return it->second;
        }
        private:
        std::mutex _mutex;
        std::unordered_map<std::string,Channel::ptr> channels;
    };
}
#endif
